package com.exd.kafka.clickhouse;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import com.exd.kafka.service.ProducerService;
import com.exd.kafka.task.JsonTool;

@Service
public class ClickhouseServiceImpl {
	
	public final static String GAOJING_START_TIME_INDEX="gaojing_send_data_start_time";
	
	@Autowired
	private RedisTemplate<String, String> redisTemplate;
	@Autowired
	private Environment environment;
	@Autowired
	private ProducerService producerService;
	
	
	public void getSendVswData() {
		String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
		String startIndexTime = startIndexTime();
		System.out.println("---------------startTime:"+startIndexTime);
		if(startIndexTime == null || startIndexTime.trim().length() == 0)return;
		// 转换时间
		String endTime = timeCalculation(stringToDate(startIndexTime),1);
		String querySql ="SELECT vin,toString(collect_time) AS collectionTime,data FROM data_default_all WHERE  vin='LDPTEST62NE008888' AND  collect_time >= '"+startIndexTime+"' AND  collect_time < '"+endTime+"'";
		List<Map<String, Object>> queryDataList = JdbcUtil.queryData(querySql);
		if(queryDataList!=null && queryDataList.size() > 0) {
			for (int i = 0; i < queryDataList.size(); i++) {
				 Map<String, Object> map = queryDataList.get(i);
				 Date stringToDateSSS = stringToDateSSS(map.get("collectionTime").toString());
				 map.put("stringToDateSSS", stringToDateSSS.getTime());
				 producerService.sendMessageToKafka(kafkaTopic, JsonTool.ObjectToJson(map));
			}
		}
		redisTemplate.opsForValue().set(GAOJING_START_TIME_INDEX, endTime);
	}
	
	// 获取数据的开始时间
	public String startIndexTime() {
		String startTimeDataValue = redisTemplate.opsForValue().get(GAOJING_START_TIME_INDEX);
		if(startTimeDataValue == null || startTimeDataValue.trim().length() == 0) {
			String querySql ="SELECT toString(min(collect_time)) AS minTime from data_default_all";
			List<Map<String, Object>> queryData = JdbcUtil.queryData(querySql);
			Object object = queryData.get(0).get("minTime");
			String startTimeDataValueString = dateToString(stringToDate(object.toString()));
			redisTemplate.opsForValue().set(GAOJING_START_TIME_INDEX, startTimeDataValueString);
			return startTimeDataValueString;
		}else {
			return startTimeDataValue;
		}
	}
	
	public static Date stringToDate(String dateTime) {
		Date dateTimeValue = null;
		SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		try {
			dateTimeValue = simpleDateFormat.parse(dateTime);
		} catch (ParseException e) {
			e.printStackTrace();
		}
		return dateTimeValue;
	}
	
	public static Date stringToDateSSS(String dateTime) {
		Date dateTimeValue = null;
		SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
		try {
			dateTimeValue = simpleDateFormat.parse(dateTime);
		} catch (ParseException e) {
			e.printStackTrace();
		}
		return dateTimeValue;
	}
	
	public static String dateToString(Date dateTime) {
		String resultData = null;
		SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		try {
			resultData = simpleDateFormat.format(dateTime);
		} catch (Exception exception) {
			exception.printStackTrace();
		}
		return resultData;
	}
	
	public static String timeCalculation(Date dateTime,int minuteValue) {
		Calendar calendar = Calendar.getInstance();
	    calendar.setTime(dateTime);
	    calendar.add(Calendar.MINUTE, minuteValue);
	    return dateToString(calendar.getTime());
	}

}
